#include "connection.h"
#include "common/str.h"
#include "server.h"
#include "common/log.h"
#include "common/file.h"
#include "common/threadpool.h"

#include <sys/sendfile.h>


namespace evm {
    namespace http {

        HTTPConnection::HTTPConnection(int fd, HTTPServer *server)
                : fd_(fd), server_(server),
                  date_str_(server->date_str_), timer_queue_(server->timer_queue_),
                  parser_(this), wbuffer_(fd) {

            loop_ = server->get_loop();
            pool_ = server->pool_;
        }

        void HTTPConnection::start() {
            log_info("connection start");
            start_deadline_timer();

            loop_->add_event(fd_, EV_READABLE, [this](int, int) {
                do_read();
            });

            loop_->add_event(fd_, EV_WRITABLE, [this](int, int) {
                do_write();
            });
        }

        void HTTPConnection::do_read() {
            //log_info("do_read");
            cancel_deadline_timer();
            lock_guard<recursive_mutex> lock(read_mutex_);

            if (read_state_ == RECIEVE_STREAM) {
                readable_ = true;
                can_read_.notify_one();
                return;
            }

            if (is_writing_) {  // error
                close_connection_ = true;
                return;
            }

            bool end = false;
            bool error = false;
            //log_info("do_read");
            size_t  nread = 0;
            while (true) {
                ssize_t n = read(fd_, rbuffer_ + nread, RW_BYTES - nread);
                //log_info("do_read, {}", n);
                if (n == -1) {
                    //log_info("do_read");
                    if (errno == EINTR)
                        continue;
                    if (errno != EAGAIN) {
                        end = true;
                        error = true;
                        log_error("http connection read: {}", strerror(errno));
                    }
                    log_info("do_read");
                    break;
                }
                else if (n == 0) {
                    end = true;
                    break;
                }
                /*else
                    log_info("do_read, {}", string(rbuffer_, n));*/
                nread += n;
                if (nread == static_cast<size_t>(RW_BYTES))
                    break;
            }

            //log_info("do_read");
            if (!error && nread > 0) {
                size_t nparsed{};
                try {
                    nparsed = parser_.execute(rbuffer_, static_cast<size_t>(nread));
                } catch (const exception& e) {
                    log_error(e.what());
                    read_state_ = CLIENT_ERROR;
                }
                //log_info("do_read: read_state_={}, nread={}, nparsed={}", read_state_, nread, nparsed);
                switch (read_state_) {
                    case NORMAL:
                        if (nparsed == nread)
                            break;
                    case CLIENT_ERROR:
                    case SERVER_ERROR:
                        error = true;
                        break;
                    case FINISHED:
                        respond();
                        break;
                    case RECIEVE_STREAM: {
                        string s(rbuffer_ + nparsed + 1, nread - nparsed - 1);
                        //log_info("do_read: {}", s);
                        fr_->hold(rbuffer_ + nparsed + 1, nread - nparsed - 1);
                        break;
                    }
                    case ASYNC_HANDLE:
                        break;
                    default:
                        error = true;
                        break;
                }
            }

            //log_info("do_read");
            if (end || error) {
                parser_.end();
                destroy();
            }
            else
                start_deadline_timer();
        }

        void HTTPConnection::do_write() {
            //log_info("do_write");
            lock_guard<recursive_mutex> lock(write_mutex_);
            writeable_ = true;

            if (need_write_)
                send_stream();
        }

        /*
         * false, terminate parsing and respond; true, continue to handle body
         */
        bool HTTPConnection::handle_header() {
            // HTTP 1.1 Expect: 100-continue
            if (parser_.check_version(1, 1) && parser_.has_header("expect", "100-continue")) {
                response_ = Response(100);
                read_state_ = FINISHED;
                return false;
            }

            if (request_.has_header("Connection", "Keep-Alive"))
                response_.set_header("Connection", "Keep-Alive");  // follow client header

            if (parser_.check_version(1, 0) && !request_.has_header("connection")) { // HTTP/1.0
                log_info("close connection");
                close_connection_ = true;
            }
            else if (parser_.check_version(1, 1)) { // HTTP/1.1
                if (request_.has_header("connection", "close")) {
                    log_info("close connection");
                    close_connection_ = true;
                }
                if (!request_.has_header("host")) {
                    log_info("no host");
                    response_ = Response::bad_request("no host");
                    read_state_ = FINISHED;
                    return false;
                }
            }

            request_.content_length = 0;
            if (request_.has_header("Content-Length")) {
                try {
                    request_.content_length = stoul(request_.get_header("Content-Length"));
                }
                catch (const invalid_argument &e) {
                    response_ = Response::bad_request(e.what());
                    read_state_ = CLIENT_ERROR;
                    return false;
                }
            }

            if (request_.has_header("Content-Type", "application/octet-stream")) {
                fr_ = make_shared<FileReciever>(request_.content_length);

                request_.has_file = true;
                read_state_ = RECIEVE_STREAM;

                pool_->post([](void* data) {
                    auto conn = static_cast<HTTPConnection*>(data);
                    conn->server_->handle(conn->request_, conn->response_);

                    if (!conn->request_.file_name.empty()) {
                        try {
                            conn->recv_file_ = make_shared<File>(conn->request_.file_name, O_WRONLY | O_CREAT);
                        }
                        catch (const exception &e) {
                            log_error(e.what());
                            conn->response_ = Response::internal_error(e.what());
                            /*read_state_ = SERVER_ERROR;*/
                            conn->close_connection_ = true;
                            return conn->respond();
                        }
                        conn->fr_->set_file(conn->recv_file_);
                        conn->recieve_stream(true);
                    }
                    else {
                        conn->response_ = Response::bad_request("reject octet stream");
                        conn->close_connection_ = true;
                        return conn->respond();
                    }
                }, this);

                return false;
            }

            return true;
        }

        void HTTPConnection::recieve_stream(bool begin) {
            //log_info("recieve_stream");
            unique_lock<recursive_mutex> lock(read_mutex_);

            cancel_deadline_timer();

            int ret;

            if (begin) {
                ret = fr_->recieve();
            }
            else {
                if (!readable_)
                    can_read_.wait(lock, [this] { return readable_.load(); });

                auto readFull = [](int fd, void* buf, size_t count) {
                    char *b = static_cast<char *>(buf);
                    ssize_t totalBytes = 0;
                    ssize_t r;
                    do {
                        r = read(fd, b, count);
                        if (r == -1) {
                            if (errno == EINTR) {
                                continue;
                            }
                            else if (errno == EAGAIN)
                                break;
                            return r;
                        }

                        totalBytes += r;
                        b += r;
                        count -= r;
                    } while (r != 0 && count);  // 0 means EOF

                    return totalBytes;
                };

                ssize_t n = readFull(fd_, rbuffer_, RW_BYTES);
                if (n == -1) {
                    if (errno == EAGAIN) {  // read EAGAIN
                        readable_ = false;
                        pool_->post([](void* data) {
                            auto conn = static_cast<HTTPConnection*>(data);
                            conn->recieve_stream();
                        }, this);
                    }
                    else {
                        log_error("http connection read: {}", strerror(errno));
                        destroy();
                    }
                    return;
                }
                ret = fr_->recieve(rbuffer_, static_cast<size_t>(n));
            }

            //log_info("ret: {}", ret);

            switch(ret) {
                case FR_REDUNDANT:
                    response_ = Response::bad_request("content-length error");
                    break;
                case FR_ERROR:
                    response_ = Response::internal_error("write file error");
                    break;
                case FR_FINISHED:
                    if (request_.file_handler) {
                        try {
                            request_.file_handler();
                        } catch (const exception& e) {
                            auto msg = fmt::format("file handling error: {}", e.what());
                            log_error(msg);
                            response_ = Response::internal_error(msg);
                        }
                    }
                    break;
                case FR_NEEDMORE:
                    pool_->post([](void* data) {
                        auto conn = static_cast<HTTPConnection*>(data);
                        conn->recieve_stream();
                    }, this);
                    return;
                default:
                    throw runtime_error("switch branch error");
            }

            respond();

            start_deadline_timer();
        }

        void HTTPConnection::handle_body() {
            /*if (request_.has_file) {
                recv_file_.reset();  // close file
                if (request_.file_handler)
                    request_.file_handler();
                request_.file_handler = nullptr;
            }
            else {*/

            if (request_.has_header("Content-Type", "application/json")) {
                try {
                    request_.json = Json::parse(request_.body);
                } catch (const logic_error &e) {
                    /*log_error("json parse error: {}", e.what());
                    response_ = Response::bad_request(e.what());
                    read_state_ = CLIENT_ERROR;
                    return;*/
                } catch (...) {
                    throw;
                }
            }

            //log_info("response file_name: {}", response_.file_name.empty());
            //log_info("{}", (size_t)this);
            //ptrdiff_t this_ = (ptrdiff_t)this;
            //log_info("{}", this_);

            pool_->post([] (void* data) {
                auto conn = static_cast<HTTPConnection*>(data);
                conn->is_writing_ = true;
                //log_info("is_writing");

                conn->server_->handle(conn->request_, conn->response_);

                if (!conn->response_.file_name.empty()) {
                    try {
                        conn->send_file_ = make_shared<File>(conn->response_.file_name, O_RDONLY);
                        conn->response_.set_header("Content-Type", "application/octet-stream");
                        conn->offset_ = 0;
                        conn->cancel_deadline_timer();
                    } catch (const exception &e) {
                        conn->response_ = Response::internal_error(e.what());
                    }
                }

                conn->respond();
            }, this);

            read_state_ = ASYNC_HANDLE;
        }

        void HTTPConnection::respond() {
            //log_info("respond");
            is_writing_ = true;

            static const unordered_map<int, string> statuses = {
                    {100, "HTTP/1.1 100 Continue\r\n"},

                    {200, "HTTP/1.1 200 OK\r\n"},
                    {201, "HTTP/1.1 201 Created\r\n"},
                    {202, "HTTP/1.1 202 Accepted\r\n"},
                    {204, "HTTP/1.1 204 No Content\r\n"},

                    {300, "HTTP/1.1 300 Multiple Choices\r\n"},
                    {301, "HTTP/1.1 301 Moved Permanently\r\n"},
                    {302, "HTTP/1.1 302 Moved Temporarily\r\n"},
                    {304, "HTTP/1.1 304 Not Modified\r\n"},

                    {400, "HTTP/1.1 400 Bad Request\r\n"},
                    {401, "HTTP/1.1 401 Unauthorized\r\n"},
                    {403, "HTTP/1.1 403 Forbidden\r\n"},
                    {404, "HTTP/1.1 404 Not Found\r\n"},

                    {500, "HTTP/1.1 500 Internal Server Error\r\n"},
                    {501, "HTTP/1.1 501 Not Implemented\r\n"},
                    {502, "HTTP/1.1 502 Bad Gateway\r\n"},
                    {503, "HTTP/1.1 503 Service Unavailable\r\n"}
            };

            static const string sep = ": ";
            static const string crlf = "\r\n";

            auto add_res = [this](const string &s) {
                wbuffer_.push_back(s);
            };

            wbuffer_.clear();
            // wbuffer_.reserve(4 * (response_.headers.size() + 5));

            // Maybe status code not supported
            if (!statuses.count(response_.code))
                response_.code = 500;

            add_res(statuses.find(response_.code)->second);

            for (auto &header: response_.headers) {
                add_res(header.first);
                add_res(sep);
                add_res(header.second);
                add_res(crlf);
            }

            string body = response_.json.is_null() ? std::move(response_.body) : response_.json.dump();

            if (!response_.has_header("content-length")) {
                static const string content_length_field = "Content-Length: ";
                string content_length;
                if (send_file_)
                    content_length = to_string(send_file_->size());
                else
                    content_length = to_string(body.size());
                add_res(content_length_field);
                add_res(content_length);
                add_res(crlf);
                //log_info("content_length: {}", content_length);
            }

            if (!response_.has_header("server")) {
                static const string server_field = "Server: ";
                add_res(server_field);
                add_res(server_->server_name);
                add_res(crlf);
            }

            if (!response_.has_header("data")) {
                static const string date_field = "Date: ";
                add_res(date_field);
                add_res(date_str_());
                add_res(crlf);
            }

            add_res(crlf);

            if (!send_file_)
                add_res(body);

            response_.clear();

            //log_info("respond");
            send_stream();
            //log_info("respond");
        }

        void HTTPConnection::send_stream() {
            //log_info("send_stream");
            lock_guard<recursive_mutex> lock(write_mutex_);
            //log_info("send_stream");

            need_write_ = true;

            if (!writeable_)
                return;

            bool error = false;

            //log_info("send_stream");

            if (!wbuffer_.write())
                error = true;
            else {
                //log_info("send_stream");
                if (!wbuffer_.empty()) {  // has not finished writing
                    writeable_ = false;
                    return;
                }

                if (send_file_) {  // send file
                    int sent = 0;
                    while (true) {
                        int n = sendfile(fd_, send_file_->fd(), &offset_, RW_BYTES);
                        if (n == -1) {
                            if (errno == EINTR)
                                continue;
                            if (errno == EAGAIN) {  // has not finished writing
                                writeable_ = false;
                                return;
                            }
                            error = true;
                            break;
                        }
                        if (n == 0) {  // reached eof
                            send_file_.reset();  // close file
                            need_write_ = false;
                            //log_info("sendfile end");
                            start_deadline_timer();
                            break;
                        }
                        //log_info("sendfile n: {}", n);
                        sent += n;
                        if (sent == RW_BYTES) {
                            // continue to send in the next task
                            pool_->post([](void* data) {
                                auto conn = static_cast<HTTPConnection*>(data);
                                conn->send_stream();
                            }, this);
                            return;
                        }
                    }
                }
                else
                    need_write_ = false;

                //log_info("send_stream");
            }

            is_writing_ = false;
            read_state_ = NORMAL;

            if (error || close_connection_)
                destroy();

            //log_info("send_stream");
        }

        void HTTPConnection::destroy() {
            cancel_deadline_timer();
            //log_info("delete this, {}", reinterpret_cast<uintptr_t>(this));
            loop_->rmv_event(fd_, EV_READABLE | EV_WRITABLE);
            //log_info("delete this, {}", reinterpret_cast<uintptr_t>(this));
            close(fd_);
            // here delete itself is safe, see
            log_info("delete this, {}", reinterpret_cast<uintptr_t>(this));
            delete this;
        }

        HTTPConnection::~HTTPConnection() {
            log_info("http connection deconstrured");
        }

        void HTTPConnection::start_deadline_timer() {
            log_info("start_deadline_timer, {}", reinterpret_cast<uintptr_t>(this));
            cancel_deadline_timer();
            timer_key_ = timer_queue_.put([this] {
                close(fd_);
                log_info("timer_queue_, {}", reinterpret_cast<uintptr_t>(this));
                destroy();
            });
        }

        void HTTPConnection::cancel_deadline_timer() {
            log_info("cancel_deadline_timer, {}", reinterpret_cast<uintptr_t>(this));
            timer_queue_.cancel(timer_key_);
        }

    }
}  // namespaces
